package com.feiyue.learning.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * com.feiyue.learning.curator 客户端操作 zk
 * @author  feiyue
 * @date  2020/5/3
 */
public class CuratorDemo {
    private static final String CONNECTION = "127.0.0.1:2181";
    private static final String PATH = "/test";
    private static final String DATA = "data";

    public static void main(String[] args) {

        // 重试策略, 第一次等待 1s, 第二次等待 2s, 第三次等待 4s
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client1 = CuratorFrameworkFactory.newClient(CONNECTION, retryPolicy);


        // 使用建造者模式创建 CuratorFramework
        CuratorFramework client2 = CuratorFrameworkFactory.builder()
                .connectString(CONNECTION)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(1000)
                .sessionTimeoutMs(1000)
                .build();

        // 启动客户端，连接服务器
        client1.start();

        // 1 创建临时有序节点
        try {
            client1.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
                    .forPath(PATH, DATA.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            client1.close();
        }

        // 2 获取节点数据并更新
        try {
            Stat stat = client1.checkExists().forPath(PATH);

            if (stat != null) {
                byte[] payload = client1.getData().forPath(PATH);
                String data = new String(payload, "UTF-8");
                System.out.println("get data from zk " + PATH + ": " + data);

                // 获取子节点列表
                List<String> children = client1.getChildren().forPath(PATH);
                System.out.println("子节点数：" + children.size());

                // 同步更新子节点
                client1.setData().forPath(PATH, DATA.concat("_1").getBytes());

                // 异步更新子节点
                AsyncCallback.StringCallback callback = new AsyncCallback.StringCallback() {
                    @Override
                    public void processResult(int i, String s, Object o, String s1) {
                        System.out.println("i:" + i + ", s:" + s + ", o" + o + ", s1:" + s1);
                    }
                };
                client1.setData().inBackground(callback).forPath(PATH, DATA.concat("_2").getBytes());
                TimeUnit.SECONDS.sleep(1);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            client1.close();
        }

        // 事务执行操作
        try {
            client1.inTransaction()
                    .create()
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                    .forPath(PATH.concat("/test"), DATA.getBytes())
                    .and()
                    .setData().forPath(PATH, DATA.concat("_3").getBytes())
                    .and()
                    .commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            client1.close();
        }

        // 监听节点变化
        try {
            NodeCache nodeCache = new NodeCache(client1, PATH, false);
            nodeCache.start(true);
            nodeCache.getListenable().addListener(() -> {
                System.out.println("节点数据发生变化");
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            client1.close();
        }
    }
}
